/*
 * Copyright 2021-2022 Open Kunlun Technology <https://www.openkunlun.io>
 */

package io.openkunlun.scaladsl.server

import akka.actor.{ ExtendedActorSystem, Extension, ExtensionId, ExtensionIdProvider }
import akka.grpc.scaladsl.Metadata
import io.openkunlun.scaladsl.v1.{ BindingEventRequest, BindingEventResponse, ListBindingResponse }

import scala.concurrent.{ ExecutionContext, Future }

/**
 * @author ericxin.
 */
private[server] object DaprBinding extends ExtensionId[DaprBinding] with ExtensionIdProvider {
  override def lookup: ExtensionId[DaprBinding] = DaprBinding
  override def createExtension(system: ExtendedActorSystem) = new DaprBinding(system)
}
private[server] class DaprBinding(system: ExtendedActorSystem) extends Extension {

  private val lock = new Object()
  @volatile private var handlers = Map.empty[String, BindingHandler]

  def addHandler(handler: BindingHandler): Unit = {
    lock.synchronized {
      handlers = handlers + (handler.name -> handler)
    }
  }

  def getHandler(name: String): Option[BindingHandler] = {
    handlers.get(name)
  }

  def listBinding(metadata: Metadata)(implicit ec: ExecutionContext): Future[ListBindingResponse] = {
    Future.successful(ListBindingResponse(handlers.keys.toSeq))
  }

  def onBindingEvent(in: BindingEventRequest, metadata: Metadata)(implicit ec: ExecutionContext): Future[BindingEventResponse] = {
    handlers
      .get(in.name)
      .map(it => it.handle(in, metadata))
      .getOrElse(Future.failed(new BindingHandlerNotFoundException(in.name)))
  }

  def isHealthy: Boolean = handlers.nonEmpty
}
